package com.gxlevi.batch_process.task

import com.gxlevi.batch_process.bean.{OrderRecord, OrderRecordWide}
import org.apache.commons.lang3.time.FastDateFormat
import org.apache.flink.api.scala.DataSet
import org.apache.flink.api.scala._

object PreprocessTask {
  def process(orderRecorderDataSet: DataSet[OrderRecord]) = {
    orderRecorderDataSet.map {
      orderRecord => {
        OrderRecordWide(
          orderRecord.benefitAmount,
          orderRecord.orderAmount,
          orderRecord.payAmount,
          orderRecord.activityNum,
          orderRecord.createTime,
          orderRecord.merchantId,
          orderRecord.orderId,
          orderRecord.payTime,
          orderRecord.payMethod,
          orderRecord.voucherAmount,
          orderRecord.commodityId,
          orderRecord.userId,
          formatTime(orderRecord.createTime, "yyyyMMdd"),
          formatTime(orderRecord.createTime, "yyyyMM"),
          formatTime(orderRecord.createTime, "yyyy")
        )
      }
    }
  }

  def formatTime(date: String, format: String) = {
    val dateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss")
    val timeStamp = dateFormat.parse(date).getTime
    FastDateFormat.getInstance(format).format(timeStamp)
  }
}
